package com.doit.demo.day05;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * @DATE 2022/2/19/16:22
 * @Author MDK
 * @Version 2021.2.2
 *
 *  assignTimeStampAndWaterMarks方法也是一个transformation,不会改变数据的样式,仅仅会提取数据中的EventTime,然后生成WaterMark发送到下游
 *  assignTimeStampAndWaterMarks方法返回的DataStream并行度 = 调用该方法的DataStream的并行度
 *
 **/
public class EventTimeTumblingWindowDemo3 {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> lines = env.socketTextStream("linux01", 7777);

        //提取数据中的EventTime,按照数据中的时间划分窗口
        //该方法是用于提取数据中的时间,不会改变数据原来的格式
        //WaterMark是一种特殊的消息,用于提取EventTime的算子,可以向下游发送给窗口对应的task
        //waterMark = 每个分区中最大的EventTime - 延迟时间
        //窗口触发的时机: WaterMark >= 窗口的结束时间
        SingleOutputStreamOperator<Tuple3<Long, String, Integer>> tp = lines.map(new MapFunction<String, Tuple3<Long, String, Integer>>() {
            @Override
            public Tuple3<Long, String, Integer> map(String line) throws Exception {
                String[] fields = line.split(",");
                long timeStamp = Long.parseLong(fields[0]);
                String word = fields[1];
                int count = Integer.parseInt(fields[2]);
                return Tuple3.of(timeStamp, word, count);
            }
        });

        //调用assignTimeStampAndWaterMarks方法生成Watermarks
        SingleOutputStreamOperator<Tuple3<Long, String, Integer>> tpAndWatermarks = tp.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Tuple3<Long, String, Integer>>(Time.seconds(2)) {
            @Override
            public long extractTimestamp(Tuple3<Long, String, Integer> tp) {
                return tp.f0;
            }
        });

        //
    }
}
